package com.webull.information.center.carwler.job;

import java.util.List;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.lang3.ArrayUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.webull.information.center.carwler.common.util.aws.s3.AwsS3ClientUtil;
import com.webull.information.center.carwler.common.util.aws.s3.AwsUsS3ClientUtil;
import com.webull.information.center.carwler.common.util.aws.s3.S3WullObject;
import com.webull.information.client.model.News;
import com.webull.information.client.service.InfoNewsService;

/**
 * 新闻S3文件同步
 * 
 * @author shimingjun
 * @date 2016年12月5日 上午10:34:53
 * @version 1.0
 * @since JDK 1.8
 */
@Component
public class News_US_synch_Job implements SimpleJob {
	protected final Logger logger = LogManager.getLogger(getClass());
	private @Autowired InfoNewsService infoNewsService;
	private @Autowired AwsS3ClientUtil awsS3ClientUtil;
	private @Autowired AwsUsS3ClientUtil awsUsS3ClientUtil;
	private @Autowired @Qualifier("executor_s3") ThreadPoolTaskExecutor threadPoolTaskExecutor;

	/**
	 * 历史已同步完，暂时关闭job
	 * 
	 * @since 0.1.0
	 */

	@Override
	public void execute(ShardingContext context) {
		int sharding = Optional.ofNullable(context).map(v -> v.getShardingItem()).orElse(-1);
		int totalCount = Optional.ofNullable(context).map(v -> v.getShardingTotalCount()).orElse(-1);

		logger.info("News_US_synch_Job:" + ArrayUtils.toString(sharding) + ",totalCount:" + totalCount);

		Integer count1 = 0;
		Long lastNewsId = null;
		do {
			List<News> lists = infoNewsService.queryPaging(lastNewsId, 100);
			if (lists == null || lists.isEmpty()) {
				count1 = 0;
			} else {
				count1 = lists.size();
				lastNewsId = lists.get(lists.size() - 1).getId();
				final CountDownLatch latch = new CountDownLatch(count1);
				final AtomicInteger at = new AtomicInteger(0);
				for_: for (News news : lists) {
					if (news.getSyncUs() != null && news.getSyncUs()) {// 已同步
						latch.countDown();
						continue for_;
					}

					if (at.incrementAndGet() % 40 == 0) {
						try {
							Thread.sleep(2000);
						} catch (Exception e) {
						}
					}

					threadPoolTaskExecutor.execute(new Runnable() {
						@Override
						public void run() {
							try {
								handler(news);
							} finally {
								latch.countDown();
							}
						}
					});

				}
				try {
					latch.await();
				} catch (InterruptedException e) {
				}
			}
		} while (count1 != null && count1 > 0);

		logger.info("news_ReutersGuide_Job div shardingRegionIds:{}; ");
	}



	private void handler(News news) {
		try {
			String key = news.getStoreShard() + "/" + news.getUrlMd5() + "." + news.getStoreFormat();
			S3WullObject s3Object = awsS3ClientUtil.downloadNewsFile(key);
			awsUsS3ClientUtil.uploadNewsFile(key, s3Object);
			/**
			 * 标记
			 */
			News news2 = new News();
			news2.setSyncUs(true);
			news2.setId(news.getId());
			infoNewsService.update(news2);
		} catch (Exception e) {
			logger.error("News_US_synch_Job handler NewsId {}, error{}", news.getId(), e);
		}

	}

}
